package com.ling.rocketmq.consumer;


import com.ling.rocketmq.listener.LingMQMessageListener;
import com.ling.rocketmq.model.ConsumerMethodModel;
import com.ling.rocketmq.properties.ConsumerProperties;
import com.ling.rocketmq.properties.RocketmqSrvProperties;
import com.ling.rocketmq.util.NameSrvAddrUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.Map;
import java.util.concurrent.ConcurrentMap;

public class ConsumerFactory implements ApplicationContextAware {
    private final static Logger log = LoggerFactory.getLogger(ConsumerFactory.class);

    private ApplicationContext applicationContext;

    private final RocketmqSrvProperties srvProperties;

    private final ConsumerProperties consumerProperties;

    private ConcurrentMap<String, ConsumerMethodModel> consumerMethodModelMap;

    public ConsumerFactory(RocketmqSrvProperties srvProperties, ConsumerProperties consumerProperties) {
        this.srvProperties = srvProperties;
        this.consumerProperties = consumerProperties;

    }

    public void start() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerProperties.getGroupName());
        consumer.setNamesrvAddr(NameSrvAddrUtil.nameSrvAddrBuild(srvProperties.getNameSrvAddr()));
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumerProperties.getConsumeMessageBatchMaxSize());
        consumer.setMaxReconsumeTimes(consumerProperties.getMaxReconsumeTimes());
        StringBuilder stringBuffer = new StringBuilder();

        for (Map.Entry<String, ConsumerMethodModel> listeners : consumerMethodModelMap.entrySet()) {
//            String[] topics = listeners.getValue().annotations.topics();
//            for (String topic : topics) {
//                consumer.subscribe(topic, "*");
//                stringBuffer.append(topic).append("、");
//            }
            String topic = listeners.getValue().annotations.topic();
            consumer.subscribe(topic, "*");
            stringBuffer.append(topic).append("、");
        }
        //MessageListener
        LingMQMessageListener listener = applicationContext.getBean(LingMQMessageListener.class);
        listener.setListenerConcurrentMap(consumerMethodModelMap);
        consumer.registerMessageListener(listener);
        consumer.start();
        log.info("rocketmq consumer start success !!! subscribe topics {}", stringBuffer.toString());


    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public void setConsumerMethodModelMap(ConcurrentMap<String, ConsumerMethodModel> consumerMethodModelMap) {
        this.consumerMethodModelMap = consumerMethodModelMap;
    }
}

